package com.bailiny.wordcount;

import com.bailiny.common.CommonConstants;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * @author Yang Xu
 * @date 2020/1/15 9:10
 * @description: MapReduce编程的八个步骤：
 * <p>
 * map阶段的两个步骤
 * 1、第一步：读取文件，解析成key,value对  这里的key，value对指代的是k1  v1
 * 2、第二步：自定义map逻辑，接收第一步读取的k1,v1 ，转换成新的k2,v2输出
 * <p>
 * shuffle阶段四个步骤
 * 3、第三步：分区  相同key的的数据发送到同一个reduce里面去，形成一个集合  这里的key指代的是k2
 * 4、第四步：排序  对我们的数据进行字典顺序的排列
 * 5、第五步：规约  主要是在map端对数据做一次聚合，减少我们输出的k2的数据量
 * 6、第六步：分组   将相同的数据发送到同一组里面去调用一次reduce逻辑
 * <p>
 * reduce阶段的两个步骤
 * 7、第七步：自定义reduce逻辑，接收k2  v2  转换成新的k3  v3 进行输出
 * 8、第八步：输出  将我们reduce处理完成之后的数据进行输出
 */
public class JobMain extends Configured implements Tool {

    @Override
    public int run(String[] args) throws Exception {
        // 获取一个job对象，通过job对象将8个步骤（即8个Class）组织到一起，提交给yarn运行
        Job job = Job.getInstance(super.getConf(), "wordCountJob");

        // 如果要打包放到服务器运行需加上这行代码，本地运行不需要
        // job.setJarByClass(JobMain.class);

        // 第一步：读取文件，解析成key，value对，即k1，v1
        job.setInputFormatClass(TextInputFormat.class);
        // 读取该路径下所有的文件
        // 集群运行模式，输出到hdfs
        // TextInputFormat.addInputPath(job, new Path(CommonConstants.HDFS_ROOT_URL + "/wordcount"));
        // 本地运行模式
        TextInputFormat.addInputPath(job, new Path(CommonConstants.LOCAL_WORD_COUNT_ROOT_DIR + "input"));
        // 第二步： 自定义map逻辑，接收第一步的k1 v1转换成新的 k2 v2进行输出
        job.setMapperClass(WordCountMapper.class);
        // 设置k2 v2的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        /**
         * 第三步到第6步
         *  分区 相同key的value发送到同一个reduce里面去，形成一个集合
         *  排序
         *  规约
         *  分组
         *  都省略
         */

        // 第七步：设置reduce类，接收key2 v2输出k3 v3
        job.setReducerClass(WordCountReducer.class);
        //设置我们reduce阶段完成之后的输出类型,设置k3 v3的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        // 第八步：设置输出类 outputformat
        job.setOutputFormatClass(TextOutputFormat.class);
        // 集群模式
        // TextOutputFormat.setOutputPath(job, new Path(CommonConstants.HDFS_ROOT_URL + "/wordcountout"));
        // 使用本地运行模式
        TextOutputFormat.setOutputPath(job, new Path(CommonConstants.LOCAL_WORD_COUNT_ROOT_DIR + "output1"));
        // 提交任务
        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {

        // 提交job任务，运行后会返回状态码，如果为0表示成功
        int run = ToolRunner.run(new Configuration(), new JobMain(), args);

        System.exit(run);
    }


}
